package org.xiangxji.util.framework.threadpool;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 
 * 扩展线程池基本实现， 在每个任务结束后需要减去任务计数器的值
 * 
 * @author xiangxji
 */
public class JobThreadPoolExecutor extends ThreadPoolExecutor {

	private JobDispatcher jobDispatcher;

	public JobThreadPoolExecutor(int maximumPoolSize, long keepAliveTime, TimeUnit unit,
			BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
			JobDispatcher jobDispatcher) {
		super(maximumPoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
		this.jobDispatcher = jobDispatcher;
	}

	@Override
	protected void afterExecute(Runnable r, Throwable t) {
		if (r instanceof Job) {
			String key = ((Job) r).getKey();
			jobDispatcher.getTotalCounter().decrementAndGet();

			if (jobDispatcher.getCounterPool().size() == 0) {
				jobDispatcher.getDefaultCounter().decrementAndGet();
			} else {
				AtomicInteger counter = jobDispatcher.getCounterPool().get(key);
				if (counter != null) {
					if (counter.decrementAndGet() < 0) { // leave mode (use default)
						counter.incrementAndGet();
						jobDispatcher.getDefaultCounter().decrementAndGet();
					} else {
						Integer size = jobDispatcher.getThresholdPool().get(key);
						if (size < 0) { // limit mode (use default)
							//counter.decrementAndGet();
							jobDispatcher.getDefaultCounter().decrementAndGet();
						} else { // leave mode (use itself)
							// nothing to do
						}
					}
				} else {
					jobDispatcher.getDefaultCounter().decrementAndGet();
				}
			}
		}
	}

}
